use async_trait::async_trait;
use fluviox_shared::{Result, FluvioXError, models::DataRecord};
use crate::traits::{DestinationConnector, ConnectorConfig, ConnectorStatus, ConnectorMetrics};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::path::PathBuf;
use tokio::fs::{File, OpenOptions};
use tokio::io::{AsyncWriteExt, BufWriter};
use chrono::Utc;
use tracing::{info, error};

/// 文件目标连接器配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileDestinationConfig {
    /// 输出文件路径
    pub path: String,
    /// 文件格式
    pub format: FileOutputFormat,
    /// 写入模式
    pub write_mode: FileWriteMode,
    /// 是否追加到文件末尾
    pub append: bool,
    /// 批次大小
    pub batch_size: u32,
    /// 文件编码
    pub encoding: String,
    /// 是否创建目录
    pub create_dirs: bool,
    /// 文件轮转配置
    pub rotation: Option<FileRotationConfig>,
}

/// 文件输出格式
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum FileOutputFormat {
    /// JSON Lines格式（每行一个JSON对象）
    JsonLines,
    /// 标准JSON格式（数组）
    Json,
    /// CSV格式
    Csv,
    /// 纯文本格式
    Text,
}

/// 文件写入模式
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum FileWriteMode {
    /// 覆盖模式
    Overwrite,
    /// 追加模式
    Append,
    /// 创建新文件（如果存在则失败）
    CreateNew,
}

/// 文件轮转配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileRotationConfig {
    /// 最大文件大小（字节）
    pub max_size: u64,
    /// 最大文件数量
    pub max_files: u32,
    /// 轮转策略
    pub strategy: RotationStrategy,
}

/// 轮转策略
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum RotationStrategy {
    /// 按大小轮转
    Size,
    /// 按时间轮转
    Time,
    /// 按记录数轮转
    Count,
}

impl ConnectorConfig for FileDestinationConfig {
    fn validate(&self) -> Result<()> {
        if self.path.is_empty() {
            return Err(FluvioXError::Validation("文件路径不能为空".to_string()));
        }

        if self.batch_size == 0 {
            return Err(FluvioXError::Validation("批次大小必须大于0".to_string()));
        }

        // 验证编码格式
        if !["utf-8", "gbk", "gb2312", "ascii"].contains(&self.encoding.to_lowercase().as_str()) {
            return Err(FluvioXError::Validation("不支持的编码格式".to_string()));
        }

        Ok(())
    }

    fn to_json(&self) -> Result<Value> {
        Ok(serde_json::to_value(self)?)
    }

    fn from_json(value: Value) -> Result<Self> {
        Ok(serde_json::from_value(value)?)
    }
}

/// 文件目标连接器
pub struct FileDestinationConnector {
    config: Option<FileDestinationConfig>,
    status: ConnectorStatus,
    metrics: ConnectorMetrics,
    current_file: Option<BufWriter<File>>,
    current_file_size: u64,
    records_written: u64,
}

impl FileDestinationConnector {
    pub fn new() -> Self {
        Self {
            config: None,
            status: ConnectorStatus::Uninitialized,
            metrics: ConnectorMetrics::default(),
            current_file: None,
            current_file_size: 0,
            records_written: 0,
        }
    }

    /// 创建输出文件
    async fn create_file(&mut self, config: &FileDestinationConfig) -> Result<BufWriter<File>> {
        let path = PathBuf::from(&config.path);

        // 创建目录（如果需要）
        if config.create_dirs {
            if let Some(parent) = path.parent() {
                tokio::fs::create_dir_all(parent).await
                    .map_err(|e| FluvioXError::Io(e))?;
            }
        }

        // 根据写入模式打开文件
        let file = match config.write_mode {
            FileWriteMode::Overwrite => {
                File::create(&path).await
                    .map_err(|e| FluvioXError::Io(e))?
            },
            FileWriteMode::Append => {
                OpenOptions::new()
                    .create(true)
                    .append(true)
                    .open(&path)
                    .await
                    .map_err(|e| FluvioXError::Io(e))?
            },
            FileWriteMode::CreateNew => {
                OpenOptions::new()
                    .create_new(true)
                    .write(true)
                    .open(&path)
                    .await
                    .map_err(|e| FluvioXError::Io(e))?
            },
        };

        Ok(BufWriter::new(file))
    }

    /// 格式化记录为字符串
    fn format_record(&self, record: &DataRecord, format: &FileOutputFormat) -> Result<String> {
        match format {
            FileOutputFormat::JsonLines => {
                Ok(serde_json::to_string(&record.data)? + "\n")
            },
            FileOutputFormat::Json => {
                // 对于JSON格式，我们需要收集所有记录然后一次性写入
                // 这里简化处理，按行写入
                Ok(serde_json::to_string(&record.data)?)
            },
            FileOutputFormat::Text => {
                match &record.data {
                    Value::String(s) => Ok(s.clone() + "\n"),
                    _ => Ok(record.data.to_string() + "\n"),
                }
            },
            FileOutputFormat::Csv => {
                // 简化的CSV格式化
                if let Value::Object(obj) = &record.data {
                    let values: Vec<String> = obj.values()
                        .map(|v| match v {
                            Value::String(s) => format!("\"{}\"", s.replace("\"", "\"\"")),
                            _ => v.to_string(),
                        })
                        .collect();
                    Ok(values.join(",") + "\n")
                } else {
                    Ok(record.data.to_string() + "\n")
                }
            },
        }
    }

    /// 检查是否需要轮转文件
    fn should_rotate(&self, config: &FileDestinationConfig) -> bool {
        if let Some(rotation) = &config.rotation {
            match rotation.strategy {
                RotationStrategy::Size => {
                    self.current_file_size >= rotation.max_size
                },
                RotationStrategy::Count => {
                    self.records_written >= rotation.max_files as u64
                },
                RotationStrategy::Time => {
                    // 时间轮转需要更复杂的逻辑，这里简化
                    false
                },
            }
        } else {
            false
        }
    }
}

#[async_trait]
impl DestinationConnector for FileDestinationConnector {
    type Config = FileDestinationConfig;

    fn name(&self) -> &str {
        "file-destination"
    }

    fn version(&self) -> &str {
        "1.0.0"
    }

    fn description(&self) -> &str {
        "文件目标连接器，支持多种格式和写入模式"
    }

    async fn initialize(&mut self, config: Self::Config) -> Result<()> {
        info!("初始化文件目标连接器");

        // 验证配置
        config.validate()?;

        // 创建初始文件
        let file = self.create_file(&config).await?;
        self.current_file = Some(file);
        self.current_file_size = 0;
        self.records_written = 0;

        self.config = Some(config);
        self.status = ConnectorStatus::Initialized;

        info!("文件目标连接器初始化完成");
        Ok(())
    }

    async fn test_connection(&self) -> Result<bool> {
        info!("测试文件连接");

        let config = self.config.as_ref()
            .ok_or_else(|| FluvioXError::Internal("连接器未初始化".to_string()))?;

        let path = PathBuf::from(&config.path);

        // 检查父目录是否存在或可创建
        if let Some(parent) = path.parent() {
            if !parent.exists() && config.create_dirs {
                match tokio::fs::create_dir_all(parent).await {
                    Ok(_) => {
                        info!("文件连接测试成功");
                        Ok(true)
                    },
                    Err(e) => {
                        error!("文件连接测试失败: {}", e);
                        Ok(false)
                    }
                }
            } else if parent.exists() {
                info!("文件连接测试成功");
                Ok(true)
            } else {
                error!("父目录不存在且未启用自动创建");
                Ok(false)
            }
        } else {
            info!("文件连接测试成功");
            Ok(true)
        }
    }

    async fn start(&mut self) -> Result<()> {
        info!("启动文件目标连接器");
        self.status = ConnectorStatus::Running;
        Ok(())
    }

    async fn stop(&mut self) -> Result<()> {
        info!("停止文件目标连接器");

        // 刷新并关闭文件
        if let Some(mut file) = self.current_file.take() {
            file.flush().await
                .map_err(|e| FluvioXError::Io(e))?;
        }

        self.status = ConnectorStatus::Stopped;
        Ok(())
    }

    async fn write_records(&mut self, records: Vec<DataRecord>) -> Result<usize> {
        let config = self.config.clone()
            .ok_or_else(|| FluvioXError::Internal("连接器未初始化".to_string()))?;

        if records.is_empty() {
            return Ok(0);
        }

        info!("写入 {} 条记录到文件", records.len());

        let mut written_count = 0;

        for record in records {
            // 检查是否需要轮转文件
            if self.should_rotate(&config) {
                // 关闭当前文件
                if let Some(mut file) = self.current_file.take() {
                    file.flush().await
                        .map_err(|e| FluvioXError::Io(e))?;
                }

                // 创建新文件（这里简化处理，实际应该生成新的文件名）
                let new_file = self.create_file(&config).await?;
                self.current_file = Some(new_file);
                self.current_file_size = 0;
                self.records_written = 0;
            }

            // 格式化记录
            let formatted = self.format_record(&record, &config.format)?;
            let data_bytes = formatted.as_bytes();

            // 写入文件
            if let Some(file) = &mut self.current_file {
                file.write_all(data_bytes).await
                    .map_err(|e| FluvioXError::Io(e))?;

                self.current_file_size += data_bytes.len() as u64;
                self.records_written += 1;
                written_count += 1;
            } else {
                return Err(FluvioXError::Internal("文件未打开".to_string()));
            }
        }

        // 刷新缓冲区
        if let Some(file) = &mut self.current_file {
            file.flush().await
                .map_err(|e| FluvioXError::Io(e))?;
        }

        // 更新指标
        self.metrics.records_processed += written_count as u64;
        self.metrics.bytes_processed += self.current_file_size;
        self.metrics.last_activity = Some(Utc::now());

        info!("成功写入 {} 条记录", written_count);
        Ok(written_count)
    }

    fn status(&self) -> ConnectorStatus {
        self.status.clone()
    }

    fn metrics(&self) -> ConnectorMetrics {
        self.metrics.clone()
    }
}
